🌎 rns.kin.earth git 🌍
Commit 929aba02821a537a260f53ed59de2e9066bd0743
Parents : 7347034
Author : Mark Qvist <bc7291552be7a58f361522990465165c>
Signature : T66BB85Valid, signed by author
Date : 2026-08-23T23:47:34+02:00
Added IFAC tests
Changes
Diff
diff --git a/tests/ifac.py b/tests/ifac.py
new file mode 100644
index 00000000..48db799d
--- /dev/null
+++ b/tests/ifac.py
@@ -0,0 +1,296 @@
+import unittest
+
+import os
+import time
+import random
+
+import RNS
+from RNS import Transport
+
+IFAC_TEST_SEED = 0x1FACAD3
+IFAC_NETNAME = b"rns-ifac-test-net"
+
+
+def rand_bytes(rng, length):
+ return bytes(rng.randrange(256) for _ in range(length))
+
+
+class FakeInterface:
+ def __init__(self, ifac_size, ifac_key):
+ self.ifac_size = ifac_size
+ self.ifac_key = ifac_key
+ self.ifac_identity = RNS.Identity.from_bytes(ifac_key)
+ self.violations = []
+
+ def ifac_violation(self, description):
+ self.violations.append(description)
+
+
+def make_interface(ifac_size):
+ ifac_key = RNS.Cryptography.hkdf(length=64, derive_from=RNS.Identity.full_hash(IFAC_NETNAME),
+ salt=RNS.Reticulum.IFAC_SALT, context=None)
+
+ return FakeInterface(ifac_size, ifac_key)
+
+
+def make_plain_packet(rng, size, pattern="random"):
+ """Generate a deterministic protocol-realistic packet: a random or
+ structured payload with the IFAC flag bit of the first header byte
+ clear (as plaintext frames always have)."""
+ if pattern == "random": raw = rand_bytes(rng, size)
+ elif pattern == "zeros": raw = bytes(size)
+ elif pattern == "ones": raw = bytes([0xFF]) * size
+ elif pattern == "seq": raw = bytes((i % 256) for i in range(size))
+ else: raise ValueError(f"Unknown pattern {pattern}")
+
+ raw = bytearray(raw)
+ raw[0] &= 0x7F
+ return bytes(raw)
+
+
+class TestIFAC(unittest.TestCase):
+
+ def test_01_outgoing_parity(self):
+ print("")
+
+ rng = random.Random(IFAC_TEST_SEED)
+
+ sizes = [8, 16, 32, 100, 500, 1064, 4096, 16384]
+ ifac_sizes = [1, 8, 16]
+ patterns = ["random", "zeros", "ones", "seq"]
+
+ verified = 0
+ for size in sizes:
+ for ifac_size in ifac_sizes:
+ iface = make_interface(ifac_size)
+ for pattern in patterns:
+ raw = make_plain_packet(rng, size, pattern)
+
+ out_new = Transport.handle_outgoing_ifac(iface, raw)
+ out_leg = Transport.handle_outgoing_ifac_legacy(iface, raw)
+
+ self.assertEqual(out_new, out_leg)
+ self.assertEqual(len(out_new), size + ifac_size)
+ verified += 1
+
+ for ifac_size in (8, 16):
+ iface = make_interface(ifac_size)
+ raw = make_plain_packet(rng, 262144, "random")
+ out_new = Transport.handle_outgoing_ifac(iface, raw)
+ out_leg = Transport.handle_outgoing_ifac_legacy(iface, raw)
+ self.assertEqual(out_new, out_leg)
+ verified += 1
+
+ print(f"IFAC outgoing parity: {verified} deterministic cases verified")
+
+ def test_02_incoming_parity(self):
+ print("")
+
+ rng = random.Random(IFAC_TEST_SEED ^ 0x1FAC)
+
+ sizes = [8, 16, 32, 100, 500, 1064, 4096, 16384]
+ ifac_sizes = [1, 8, 16]
+ patterns = ["random", "zeros", "ones", "seq"]
+
+ verified = 0
+ for size in sizes:
+ for ifac_size in ifac_sizes:
+ iface = make_interface(ifac_size)
+ for pattern in patterns:
+ raw = make_plain_packet(rng, size, pattern)
+ frame = Transport.handle_outgoing_ifac(iface, raw)
+
+ ok_new, out_new = Transport.handle_ifac(frame, iface)
+ ok_leg, out_leg = Transport.handle_ifac_legacy(frame, iface)
+
+ self.assertTrue(ok_new)
+ self.assertTrue(ok_leg)
+ self.assertEqual(out_new, out_leg)
+ verified += 1
+
+ for ifac_size in (8, 16):
+ iface = make_interface(ifac_size)
+ raw = make_plain_packet(rng, 262144, "random")
+ frame = Transport.handle_outgoing_ifac(iface, raw)
+ ok_new, out_new = Transport.handle_ifac(frame, iface)
+ ok_leg, out_leg = Transport.handle_ifac_legacy(frame, iface)
+ self.assertTrue(ok_new)
+ self.assertTrue(ok_leg)
+ self.assertEqual(out_new, out_leg)
+ verified += 1
+
+ print(f"IFAC incoming parity: {verified} deterministic cases verified")
+
+ def test_03_roundtrip(self):
+ print("")
+
+ rng = random.Random(IFAC_TEST_SEED ^ 0x0FAC)
+
+ # All four TX/RX handler combinations must recover the original packet
+ combos = [
+ (Transport.handle_outgoing_ifac, Transport.handle_ifac),
+ (Transport.handle_outgoing_ifac, Transport.handle_ifac_legacy),
+ (Transport.handle_outgoing_ifac_legacy, Transport.handle_ifac),
+ (Transport.handle_outgoing_ifac_legacy, Transport.handle_ifac_legacy),
+ ]
+
+ sizes = [16, 1064, 16384]
+ ifac_sizes = [1, 8, 16]
+
+ verified = 0
+ for tx, rx in combos:
+ for size in sizes:
+ for ifac_size in ifac_sizes:
+ iface = make_interface(ifac_size)
+ raw = make_plain_packet(rng, size, "random")
+
+ frame = tx(iface, raw)
+ ok, out = rx(frame, iface)
+
+ self.assertTrue(ok)
+ self.assertEqual(out, raw)
+ verified += 1
+
+ print(f"IFAC round-trip: {verified} deterministic TX/RX combinations verified\n All four handler pairings recover the packet exactly.")
+
+ def test_04_invariants(self):
+ print("")
+
+ rng = random.Random(IFAC_TEST_SEED ^ 0xCAFE)
+ iface = make_interface(16)
+
+ for size in (16, 1064, 16384):
+ raw = make_plain_packet(rng, size, "random")
+
+ out_new = Transport.handle_outgoing_ifac(iface, raw)
+ out_leg = Transport.handle_outgoing_ifac_legacy(iface, raw)
+
+ # IFAC flag must be set on the masked frame
+ self.assertEqual(out_new[0] & 0x80, 0x80)
+
+ # The IFAC tag region must be transmitted in the clear and
+ # identical in both implementations
+ self.assertEqual(out_new[2:2+iface.ifac_size], out_leg[2:2+iface.ifac_size])
+ self.assertEqual(out_new[2:2+iface.ifac_size], iface.ifac_identity.sign(raw)[-iface.ifac_size:])
+
+ # Second header byte must be masked identically
+ self.assertEqual(out_new[1], out_leg[1])
+
+ # The incoming handler must clear the IFAC flag in its output
+ ok, out = Transport.handle_ifac(out_new, iface)
+ self.assertTrue(ok)
+ self.assertEqual(out[0] & 0x80, 0)
+
+ print("IFAC invariants: flag set on frame, tag in clear and identical, flag cleared on input recovery")
+
+ def test_05_invalid_tags(self):
+ print("")
+
+ rng = random.Random(IFAC_TEST_SEED ^ 0xBAD)
+
+ sizes = [100, 1064, 16384]
+ ifac_sizes = [1, 8, 16]
+
+ for size in sizes:
+ for ifac_size in ifac_sizes:
+ iface = make_interface(ifac_size)
+ raw = make_plain_packet(rng, size, "random")
+ frame = bytearray(Transport.handle_outgoing_ifac(iface, raw))
+
+ corruptions = []
+ # Corrupt a masked payload byte
+ corrupted = bytearray(frame)
+ corrupted[-1] ^= 0x01
+ corruptions.append(bytes(corrupted))
+
+ # Corrupt the second header byte
+ corrupted = bytearray(frame)
+ corrupted[1] ^= 0x40
+ corruptions.append(bytes(corrupted))
+
+ # Corrupt a byte in the IFAC tag region
+ if ifac_size > 0:
+ corrupted = bytearray(frame)
+ corrupted[2 + ifac_size//2] ^= 0x01
+ corruptions.append(bytes(corrupted))
+
+ for corrupted in corruptions:
+ ok_new, out_new = Transport.handle_ifac(corrupted, iface)
+ ok_leg, out_leg = Transport.handle_ifac_legacy(corrupted, iface)
+
+ # Both implementations must behave identically on any
+ # corrupted frame, including the violation report
+ self.assertEqual(ok_new, ok_leg)
+ self.assertEqual(out_new, out_leg)
+ self.assertEqual(iface.violations[-1:], ["Invalid IFAC on packet"])
+
+ # With >= 8 bytes of tag entropy, rejection is
+ # deterministic. At ifac_size = 1 the truncated tag has
+ # only 8 bits, so a corrupted frame can statistically
+ # collide with the original truncation (1/256) - both
+ # implementations behave identically either way.
+ if ifac_size >= 8: self.assertFalse(ok_new)
+
+ print("IFAC invalid tags: behaviour and violation reporting identical "
+ "between implementations; deterministic rejection verified for "
+ "8+ byte tags")
+
+ @unittest.skipIf(os.getenv("RNS_SKIP_PERF_TESTS") != None, "Skipping performance tests")
+ def test_06_benchmark(self):
+ print("")
+
+ rng = random.Random(IFAC_TEST_SEED ^ 0xBEEF)
+
+ sizes = [1500, 16384, 262144]
+ runs = {1500: 30, 16384: 15, 262144: 5}
+
+ print("IFAC handler benchmark (optimized vs legacy):")
+ print(f"{'size':>8} {'out lgc':>11} {'out opt':>11} {'out spd':>8} | "
+ f"{'in lgc':>11} {'in new':>11} {'in spd':>8}")
+ print("-" * 84)
+
+ total_leg = total_new = 0.0
+ for size in sizes:
+ iface = make_interface(16)
+ raw = make_plain_packet(rng, size, "random")
+ frame = Transport.handle_outgoing_ifac(iface, raw)
+ n = runs[size]
+
+ # Warmup
+ Transport.handle_outgoing_ifac_legacy(iface, raw)
+ Transport.handle_outgoing_ifac(iface, raw)
+ Transport.handle_ifac_legacy(frame, iface)
+ Transport.handle_ifac(frame, iface)
+
+ t0 = time.perf_counter()
+ for _ in range(n): Transport.handle_outgoing_ifac_legacy(iface, raw)
+ out_leg = (time.perf_counter() - t0) / n
+
+ t0 = time.perf_counter()
+ for _ in range(n): Transport.handle_outgoing_ifac(iface, raw)
+ out_new = (time.perf_counter() - t0) / n
+
+ t0 = time.perf_counter()
+ for _ in range(n): Transport.handle_ifac_legacy(frame, iface)
+ in_leg = (time.perf_counter() - t0) / n
+
+ t0 = time.perf_counter()
+ for _ in range(n): Transport.handle_ifac(frame, iface)
+ in_new = (time.perf_counter() - t0) / n
+
+ total_leg += out_leg + in_leg
+ total_new += out_new + in_new
+
+ def ms(t): return f"{t*1000:.3f}m"
+
+ print(f"{size:>8} {ms(out_leg):>11} {ms(out_new):>11} {out_leg/out_new:7.1f}x | "
+ f"{ms(in_leg):>11} {ms(in_new):>11} {in_leg/in_new:7.1f}x")
+
+ print(f"\nCombined TX+RX at {sizes[-1]} bytes: legacy "
+ f"{(out_leg+in_leg)*1000:.2f} ms/pkt, optimized {(out_new+in_new)*1000:.2f} ms/pkt, "
+ f"{(out_leg+in_leg)/(out_new+in_new):.1f}x speedup")
+ print(f"Overall speedup across benchmarks: {total_leg/total_new:.1f}x")
+ print()
+
+if __name__ == "__main__":
+ unittest.main(verbosity=2)
\ No newline at end of file
Served by rngit 1.5.2 - Generated in 0.03s